package io.kiki.sba.registry.server.data_store.remoting.metaserver;

import io.kiki.sba.registry.common.model.Node.NodeType;
import io.kiki.sba.registry.common.model.Tuple;
import io.kiki.sba.registry.common.model.metaserver.inter.heartbeat.HeartBeatResponse;
import io.kiki.sba.registry.common.model.metaserver.inter.heartbeat.HeartbeatRequest;
import io.kiki.sba.registry.common.model.metaserver.nodes.DataStoreNode;
import io.kiki.sba.registry.common.model.slot.BaseSlotStatus;
import io.kiki.sba.registry.common.model.slot.SlotConfig;
import io.kiki.sba.registry.common.model.slot.Slots;
import io.kiki.sba.registry.common.model.store.Url;
import io.kiki.sba.registry.server.data_store.bootstrap.ServerConfig;
import io.kiki.sba.registry.server.data_store.multi.cluster.exchanger.RemoteDataNodeExchanger;
import io.kiki.sba.registry.server.data_store.multi.cluster.slot.MultiClusterSlotManager;
import io.kiki.sba.registry.server.data_store.remoting.DataNodeExchanger;
import io.kiki.sba.registry.server.data_store.remoting.SessionNodeExchanger;
import io.kiki.sba.registry.server.data_store.slot.SlotManager;
import io.kiki.sba.registry.server.shared.config.CommonConfig;
import io.kiki.sba.registry.server.shared.env.ServerEnv;
import io.kiki.sba.registry.server.shared.meta.AbstractMetaServerService;
import io.kiki.sba.registry.server.shared.slot.SlotsRecorder;
import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;


public class MetaServerServiceImpl extends AbstractMetaServerService<HeartBeatResponse> implements SlotsRecorder {
    protected final Logger logger = LoggerFactory.getLogger(MetaServerServiceImpl.class);
    @Autowired
    private SlotManager slotManager;

    @Autowired
    private DataNodeExchanger dataNodeExchanger;

    @Autowired
    private SessionNodeExchanger sessionNodeExchanger;

    @Autowired
    private ServerConfig serverConfig;

    @Autowired
    private CommonConfig commonConfig;

    @Autowired
    private MultiClusterSlotManager multiClusterSlotManager;

    @Autowired
    private RemoteDataNodeExchanger remoteDataNodeExchanger;

    private volatile Slots currentSlots;

    @Override
    protected long getCurrentslotsEpoch() {
        return slotManager.getSlotsEpoch();
    }

    @Override
    public int getRenewIntervalSecs() {
        return serverConfig.getSchedulerHeartbeatIntervalSecs();
    }

    @Override
    protected void handleRenewResult(HeartBeatResponse heartBeatResponse) {
        // the data/session list has updated in renewNode
        Set<String> dataServerList = getDataServerList();
        if (!CollectionUtils.isEmpty(dataServerList)) {
            dataNodeExchanger.setServerIpSet(dataServerList);
            dataNodeExchanger.notifyConnectServerAsync();
        }
        Set<String> sessionServerList = getSessionServerList();
        if (!CollectionUtils.isEmpty(sessionServerList)) {
            sessionNodeExchanger.setServerIpSet(sessionServerList);
            sessionNodeExchanger.notifyConnectServerAsync();
        }

        Map<String, Set<String>> remoteDataServers = getRemoteDataServers();
        if (!org.springframework.util.CollectionUtils.isEmpty(remoteDataServers)) {

            Set<String> dataServers = new HashSet<>(128);
            for (Set<String> servers : remoteDataServers.values()) {
                if (!CollectionUtils.isEmpty(servers)) {
                    dataServers.addAll(servers);
                }
            }
            remoteDataNodeExchanger.setServerIpSet(dataServers);
            remoteDataNodeExchanger.notifyConnectServerAsync();
        }

        if (heartBeatResponse.getSlots() != null && heartBeatResponse.getSlots().getEpoch() != Slots.INIT.getEpoch()) {
            slotManager.updateSlots(heartBeatResponse.getSlots());
        } else {
            logger.error("[handleRenewResult] slot table is {}", heartBeatResponse.getSlots() == null ? "null" : "Slots.INIT");
        }

        multiClusterSlotManager.updateSlots(heartBeatResponse.getRemoteSlotsStatusMap());
    }

    @Override
    protected HeartbeatRequest createRequest() {
        Tuple<Long, List<BaseSlotStatus>> tuple = slotManager.getslotsEpochAndStatuses(serverConfig.getDataCenter());
        final long slotsEpoch = tuple.o1;
        final List<BaseSlotStatus> slotStatuses = tuple.o2;
        HeartbeatRequest<DataStoreNode> request = new HeartbeatRequest<>(createNode(), slotsEpoch, serverConfig.getDataCenter(), System.currentTimeMillis(), SlotConfig.slotBasicInfo(), slotStatuses, multiClusterSlotManager.getSlotsEpoch()).setSlots(currentSlots);
        return request;
    }

    @Override
    protected NodeType nodeType() {
        return NodeType.data_store;
    }

    @Override
    protected String cell() {
        return commonConfig.getRegion();
    }

    private DataStoreNode createNode() {
        return new DataStoreNode(new Url(ServerEnv.IP), serverConfig.getDataCenter());
    }


    @Override
    public void record(Slots slots) {
        currentSlots = new Slots(slots.getEpoch(), slots.getSlotIdToSlotMap());
    }
}
